package org.xiong.disk.util;


import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.multipart.MultipartFile;
import org.xiong.disk.config.HdfsConfig;

import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
import java.text.SimpleDateFormat;
import java.util.*;

@Component
@Slf4j
public class HdfsUtil {

    @Autowired
    private HdfsConfig hdfsConfig;

    /**
     * 获取HDFS配置信息 配置文件优先级
     * @return
     */
    private Configuration getConfiguration() {
        Configuration configuration = new Configuration();
        configuration.set("dfs.support.append", "true");
        configuration.set("dfs.client.block.write.replace-datanode-on-failure.enable", "true");
        configuration.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
        return configuration;
    }

    /**
     * 获取HDFS文件系统对象
     * @return
     * @throws URISyntaxException
     * @throws IOException
     * @throws InterruptedException
     */
    public FileSystem getFileSystem() throws URISyntaxException, IOException, InterruptedException {
        // 客户端去操作hdfs时是有一个用户身份的，默认情况下hdfs客户端api会从jvm中获取一个参数作为自己的用户身份
        // DHADOOP_USER_NAME=hadoop
        // 也可以在构造客户端fs对象时，通过参数传递进去
        return FileSystem.get(new URI(hdfsConfig.getNameNodeUrl()), getConfiguration(), hdfsConfig.getHdfsUserName());

    }

    /**
     * 在HDFS 创建文件夹
     * @param path 传入一个路径用于创建文件夹
     * @return
     */
    public boolean mkdir(String path) throws Exception{
        FileSystem fs = null;
        boolean isOk = false;
        if (StringUtils.isEmpty(path)) {
            return false;
        }
        try {
            String newPath = hdfsConfig.getHdfsDataNode()+path;
            if (existFile(newPath)) {
                log.error("hdfs file is exists: {}", path);
                return false;
            }
            // 目标路径
            fs = getFileSystem();
            Path srcPath = new Path(newPath);
            isOk = fs.mkdirs(srcPath);
            log.info("hdfs mkdir success: {}", path);
        } catch (Exception e) {
            log.error("hdfs mkdir: { }", e);
        } finally {
            if (fs != null) {
                fs.close();
            }
        }
        return isOk;
    }
    /**
     * 判断HDFS文件是否存在
     *
     * @param path
     * @return
     * @throws Exception
     */
    public boolean existFile(String path) throws Exception {
        boolean isExists = false;
        FileSystem fs = null;
        if (StringUtils.isEmpty(path)) {
            return false;
        }
        try {
            fs = getFileSystem();
            Path srcPath = new Path(path);
            isExists = fs.exists(srcPath);
        } catch (Exception e) {
            log.error("existFile:{ }", e);
        } finally {
            if (fs != null) {
                fs.close();
            }
        }
        return isExists;
    }


    /**
     * 读取HDFS当前目录下信息
     * @param path
     * @return
     * @throws Exception
     */
    public List<Map<String, Object>> readPathInfo(String path) throws Exception {
        FileSystem fs = null;
        String newPath = hdfsConfig.getHdfsDataNode()+path;
        try {
            if (StringUtils.isEmpty(path)) {
                return null;
            }
            if (!existFile(newPath)) {
                return null;
            }
            fs = getFileSystem();
            // 目标路径
            Path srcPath = new Path(newPath);
            FileStatus[] statusList = fs.listStatus(srcPath);
            List<Map<String, Object>> list = new ArrayList<>();
            if (null != statusList && statusList.length > 0) {
                for (FileStatus fileStatus : statusList) {

                    Map<String, Object> map = new HashMap<>();
                    Path path1 = fileStatus.getPath();
                    String[] split = path1.toString().split("disk",0);
                    map.put("filePath", split[split.length-1]);
                    map.put("size",fileStatus.getLen());

                    Date date = new Date(fileStatus.getModificationTime());
                    SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
                    String newDate = dateFormat.format(date);
                    map.put("upTime",newDate);

                    if (fileStatus.isFile()) {
                        map.put("isFile",true);
                    }else {
                        map.put("isFile",false);
                    }

                    String fileName = fileStatus.getPath().getName();
                    map.put("fileName",fileName);

                    list.add(map);
                }
                return list;
            }
        } catch (Exception e) {
            log.error("hdfs readPathInfo { }", e);
        }finally {
            if (fs != null) {
                fs.close();
            }
        }
        return null;
    }


    /**
     * 读取HDFS文件列表
     * @param path
     * @return
     * @throws Exception
     */
    public List<Map<String, String>> listFile(String path) throws Exception {
        String newPath = hdfsConfig.getHdfsDataNode()+path;
        if (StringUtils.isEmpty(path)) {
            return null;
        }
        if (!existFile(newPath)) {
            return null;
        }
        FileSystem fs = null;
        try {
            fs = getFileSystem();
            // 目标路径
            Path srcPath = new Path(newPath);
            // 递归找到所有文件
            RemoteIterator<LocatedFileStatus> filesList = fs.listFiles(srcPath, true);
            List<Map<String, String>> returnList = new ArrayList<>();
            while (filesList.hasNext()) {
                LocatedFileStatus next = filesList.next();
                String fileName = next.getPath().getName();
                Path path1 = next.getPath();
                String[] filePath = path1.toString().split("/", 4);
                Map<String, String> map = new HashMap<>();
                map.put("fileName", fileName);
                map.put("filePath", "/"+filePath[filePath.length-1]);

                Date date = new Date(next.getModificationTime());
                SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
                String newDate = dateFormat.format(date);
                map.put("upTime",newDate);
                returnList.add(map);
            }
            return returnList;
        } catch (Exception e) {
            log.error("hdfs listFile { }", e);
        } finally {
            if (fs != null) {
                fs.close();
            }
        }
        return null;
    }
    /**
     * HDFS重命名文件
     *
     * @param oldName
     * @param newName
     * @return
     * @throws Exception
     */
    public boolean renameFile(String path,String oldName, String newName) throws Exception {
        if (StringUtils.isEmpty(oldName) || StringUtils.isEmpty(newName)) {
            return false;
        }
        String oldNamePath = hdfsConfig.getHdfsDataNode()+ path + oldName;
        String newNamePath = hdfsConfig.getHdfsDataNode() + path + newName;

        FileSystem fs = null;
        boolean isOk = false;
        try {
            fs = getFileSystem();

            Path oldPath = new Path(oldNamePath);

            Path newPath = new Path(newNamePath);
            isOk = fs.rename(oldPath, newPath);

            return isOk;
        } catch (Exception e) {
            log.error("hdfs renameFile:{ }", e);
        } finally {
            if (fs != null) {
                fs.close();
            }
        }
        return isOk;
    }

    /**
     * 删除HDFS文件
     *
     * @param path
     * @return
     * @throws Exception
     */
    public boolean deleteFile(String path) throws Exception {
        if (StringUtils.isEmpty(path)) {
            return false;
        }
        String newPath = hdfsConfig.getHdfsDataNode()+path;
        FileSystem fs = null;
        boolean isOk = false;
        try {
            if (!existFile(newPath)) {
                return false;
            }
            fs = getFileSystem();
            Path srcPath = new Path(newPath);
            isOk = fs.deleteOnExit(srcPath);
        } catch (Exception e) {
            log.error("hdfs deleteFile:{ }", e);
        } finally {
            if (fs != null) {
                fs.close();
            }
        }
        return isOk;
    }

    /**
     * 上传HDFS文件
     * @param file 上传文件
     * @param toPath 目标路径(全节点路径)
     * @throws Exception
     */
    public boolean uploadFile(MultipartFile file,String toPath,String fileName) throws Exception {
        boolean isOk = false;
        if (StringUtils.isEmpty(toPath) || null == file.getBytes()) {
            return isOk;
        }
        FileSystem fs = null;

        try {
            String newPath = hdfsConfig.getHdfsDataNode() + toPath + "/" + fileName;
            log.info("上传文件路径:{}",newPath);
            fs = getFileSystem();
            // 上传路径
            Path clientPath = new Path(newPath);
            // 上传io
            FSDataOutputStream out = fs.create(clientPath);

            // 目标路径
            InputStream in = file.getInputStream();
            // 使用 IOUtils.copyBytes() 进行拷贝
            IOUtils.copyBytes(in,out,1024);

            isOk = true;
            return isOk;
        } catch (Exception e) {
            log.error("hdfs uploadFile:{ }", e);
            return isOk;
        } finally {
            if (fs != null) {
                fs.close();
            }
        }
    }

    /**
     * 下载文件
     * @param path
     * @return
     * @throws IOException
     */
    public OutputStream downloadFile(String path,OutputStream os) throws IOException {
        if (StringUtils.isEmpty(path)) {
            return null;
        }
        FileSystem fs = null;
        FSDataInputStream open = null;
        InputStream in = null;
        ResponseEntity<byte[]> responseEntity = null;
        Path downloadPath = new Path(hdfsConfig.getHdfsDataNode()+path);
//        String[] split = path.split("/");
        try{
            // 获取fs文件对象
            fs = getFileSystem();
            // 打开 文件获取字节流
            open = fs.open(downloadPath);
            in = open.getWrappedStream();

//            // 创建一个相同大小的字节数组
//            byte[] bytes = new byte[in.available()];
//
//            // 将数据读取到bytes数组中
//            in.read(bytes);

            IOUtils.copyBytes(in,os,1024);

//            HttpHeaders headers = new HttpHeaders();
//            headers.add("Content-Disposition", "attachment;filename="+split[split.length-1]);
//            HttpStatus statusCode = HttpStatus.OK;


        }catch (Exception e){
            log.error("下载出错：{ }",e);
        }finally {
            IOUtils.closeStream(in);
            open.close();
            fs.close();
        }

        if ( os != null){
            return os;
        }else {
            return null;
        }
    }

}
